unit Actors;

interface

uses
  Windows, Classes, Variants;

type
  TAct = procedure (var actData: Variant) of object;

  TPooledActors = class;

  TPoolThread = class(TThread)
  private
    FActived: boolean;
    FActsCount: integer;
    FPooling: THandle;
    FPool: TPooledActors;
    procedure SetActive(const Value: boolean);
  public
    constructor Create(APool: TPooledActors);
    procedure Execute; override;

    property Active: boolean read FActived write SetActive;
  end;

  TActor = class;

  PActData = ^TActData;
  TActData = packed record
    Act: TAct;
    Data: Variant;
    Dest: TActor;
    Sender: TActor;
  end;  

  TPooledActors = class(TList)
  private
    FThreads: TThreadList;
    FThreadsCount: integer;
    FThreadsActive: integer;
    FActorHasAct: THandle;
    FCS: TRTLCriticalSection;
    FSysInfo: TSystemInfo;

    FAutoBalance: boolean;
    FActingCounter: Integer;
    FActingMaxCounter: Integer;

    procedure InitThreads(AThreadsCount: integer);
    procedure SetThreadsCount(const Value: Integer);
    procedure ActiveThreads(AThreadsCount: integer);

    procedure ActingStart(var AActData: PActData); virtual;
    procedure ActingEnd(var AActData: PActData); virtual;
  public
    constructor Create; overload;
    constructor Create(AThreadsCount: integer); overload;
    destructor Destroy; override;

    function AddActor(AActor: TActor): TActor; virtual;
    procedure RemoveActor(AActor: TActor); virtual;
    procedure DeleteActor(AActorId: integer); virtual;

    function ActorId(AActor: TActor): integer;

    procedure Dispatch(APoolThread: TPoolThread); virtual;

    property ThreadsCount: Integer read FThreadsCount write SetThreadsCount;
    property ActingMaxCounter: integer read FActingMaxCounter write FActingMaxCounter;
    property ThreadsActive: integer read FThreadsActive write FThreadsActive;
  end;

  TActor = class(TList)
  private
    FName: string;
    FActorId: integer;
    FPool: TPooledActors;
    FActLock: integer;
    FCS: TRTLCriticalSection;
  public
    constructor Create(const AName: string);
    destructor Destroy; override;

    procedure Loaded; virtual;

    function Act: boolean; virtual;

    procedure Queue(ASender: TActor; var AActData: PActData);
    function Pop(var AActData: PActData): boolean;

    procedure Send(ADestActor: TActor; AAct: TAct; var AParam: Variant); overload;
    procedure Send(ADestActor: TActor; AAct: TAct; AParam: array of Variant); overload;

    procedure Send(AAct: TAct; var AParam: Variant); overload;
    procedure Send(AAct: TAct; AParam: array of Variant); overload;

    property Name: string read FName;
  end;

implementation

function ICEInteger(var D: integer; C, E: integer) : boolean;
{$IFDEF VER150}
  var
    O: Pointer;
  begin
    O := InterlockedCompareExchange(Pointer(D), Pointer(E), Pointer(C));
    Result := Integer(O) = C;
  end;
{$ELSE}
  begin
    Result := (InterlockedCompareExchange(D,E,C) = C);
  end;
{$ENDIF}

{ TPoolThread }

constructor TPoolThread.Create(APool: TPooledActors);
begin
  FPool := APool;
  FreeOnTerminate := false;

  FActived := false;
  FPooling := CreateEvent(nil, true, false, nil);

  FActsCount := 0;
  
  inherited Create(false);
end;

procedure TPoolThread.Execute;
begin
  while not Terminated do
  begin
    WaitForSingleObject(FPooling, INFINITE);
    FPool.Dispatch(Self);
  end;
end;

procedure TPoolThread.SetActive(const Value: boolean);
begin
  FActived := Value;
  if FActived then
    SetEvent(FPooling)
  else
    ResetEvent(FPooling);  
end;

{ TPooledActors }

constructor TPooledActors.Create;
begin
  GetSystemInfo(FSysInfo);
  FThreads := TThreadList.Create;

  InitializeCriticalSection(FCS);
  InitThreads(FSysInfo.dwNumberOfProcessors);

  FActingCounter := 0;
  FAutoBalance := true;
  ActiveThreads(1);
end;

constructor TPooledActors.Create(AThreadsCount: integer);
begin
  GetSystemInfo(FSysInfo);
  FThreads := TThreadList.Create;

  InitializeCriticalSection(FCS);
  InitThreads(AThreadsCount);

  FActingCounter := 0;
  FAutoBalance := false;
  ActiveThreads(AThreadsCount);
end;

destructor TPooledActors.Destroy;
begin
  InitThreads(0);

  FThreads.Free;

  inherited;
end;

function TPooledActors.AddActor(AActor: TActor): TActor;
begin
  EnterCriticalSection(FCS);
  try
    AActor.FPool := Self;
    AActor.FActorId := Add(AActor);
  finally
    LeaveCriticalSection(FCS);
  end;
end;

procedure TPooledActors.InitThreads(AThreadsCount: integer);
var
  i: integer;
begin
  with FThreads.LockList do
    try
      for i := 0 to Count - 1 do
      begin
        TPoolThread(Items[i]).Active := true;
        TPoolThread(Items[i]).Terminate;
        TPoolThread(Items[i]).WaitFor;
        TPoolThread(Items[i]).Free;
      end;

      FThreads.Clear;

      for i := 0 to AThreadsCount - 1 do
        Add(TPoolThread.Create(Self));

      FThreadsCount := AThreadsCount;
    finally
      FThreads.UnlockList;
    end;
end;

procedure TPooledActors.Dispatch(APoolThread: TPoolThread);
var
  i: integer;
  bAct: boolean;
begin
  bAct := false;
  EnterCriticalSection(FCS);
  try
    for i := 0 to Count - 1 do
      if TActor(Items[i]).Act then
      begin
        bAct := true;
        Break;
      end;
  finally
    LeaveCriticalSection(FCS);
  end;

  if not bAct then
    Sleep(1)
  else
    Inc(APoolThread.FActsCount, 1);
end;

procedure TPooledActors.SetThreadsCount(const Value: Integer);
begin
  if FThreadsCount <> Value then
  begin
    InitThreads(Value);
    ActiveThreads(Value);
  end;
end;

procedure TPooledActors.DeleteActor(AActorId: integer);
var
  i: integer;
begin
  EnterCriticalSection(FCS);
  try
    Delete(AActorId);
    for i := AActorId to Count - 1 do
      TActor(Items[i]).FActorId := i;
  finally
    LeaveCriticalSection(FCS);
  end;
end;

procedure TPooledActors.RemoveActor(AActor: TActor);
var
  id: integer;
begin
  id := ActorId(AActor);
  if id > -1 then
    DeleteActor(id);
end;

function TPooledActors.ActorId(AActor: TActor): integer;
var
  i: integer;
begin
  Result := -1;
  EnterCriticalSection(FCS);
  try
    for i := 0 to Count - 1 do
      if Items[i] = AActor then
      begin
        Result := i;
        Break;
      end;
  finally
    LeaveCriticalSection(FCS);
  end;
end;

procedure TPooledActors.ActiveThreads(AThreadsCount: integer);
var
  i, x, ProcessorRange: integer;
  AffinityMask: DWORD;
  PoolThread: TPoolThread;
begin
  with FThreads.LockList do
    try
      ProcessorRange := FSysInfo.dwNumberOfProcessors div AThreadsCount;
      if ProcessorRange = 0 then
        ProcessorRange := 1;

      FThreadsActive := AThreadsCount;
      for i := 0 to Count - 1 do
      begin
        PoolThread := TPoolThread(Items[i]);
        PoolThread.Active := (i + 1) <= AThreadsCount;
        if PoolThread.Active then
        begin
          if ProcessorRange = 1 then
          begin
            AffinityMask := 1;
            if i > 0 then
              AffinityMask := 1 shl i;
          end else
          begin
            AffinityMask := 0;
            for x := 1 to ProcessorRange do
              AffinityMask := (AffinityMask shl 1) + (1 shl i);
          end;

          SetThreadAffinityMask(PoolThread.Handle, AffinityMask);
        end;
      end;
    finally
      FThreads.UnlockList;
    end;
end;

procedure TPooledActors.ActingStart(var AActData: PActData);
begin
  if FAutoBalance then
  begin
    EnterCriticalSection(FCS);
    try
      InterlockedIncrement(FActingCounter);
      if FActingCounter > FActingMaxCounter then
      begin
        InterlockedExchange(FActingMaxCounter, FActingCounter);
        if FActingMaxCounter > FSysInfo.dwNumberOfProcessors then
          ActiveThreads(FSysInfo.dwNumberOfProcessors)
        else if FActingMaxCounter > 1 then
          ActiveThreads(FActingMaxCounter)
        else
          ActiveThreads(1);        
      end;
    finally
      LeaveCriticalSection(FCS);
    end;
  end;
end;

procedure TPooledActors.ActingEnd(var AActData: PActData);
begin
  if FAutoBalance then
  begin
    EnterCriticalSection(FCS);
    try
      InterlockedDecrement(FActingCounter);
      if FActingCounter = 0 then
        InterlockedExchange(FActingMaxCounter, FActingCounter);
    finally
      LeaveCriticalSection(FCS);
    end;
  end;
end;

{ TActor }

constructor TActor.Create(const AName: string);
begin
  FName := AName;
  FActLock := 1;
  InitializeCriticalSection(FCS);

  Loaded;
end;

destructor TActor.Destroy;
begin
  if FPool <> nil then
    FPool.DeleteActor(FActorId);
    
  inherited;
end;

procedure TActor.Loaded;
begin
  //
end;

function TActor.Act: boolean;
var
  CurrAct: PActData;
begin
  Result := Pop(CurrAct);
  if Result then
  begin
    try
      CurrAct.Act(CurrAct.Data);
    finally
      FPool.ActingEnd(CurrAct);
    end;
    CurrAct.Data := Null;
    Dispose(CurrAct);
  end;
end;

procedure TActor.Queue(ASender: TActor; var AActData: PActData);
begin
  EnterCriticalSection(FCS);
  try
    Add(AActData);
  finally
    LeaveCriticalSection(FCS);
  end;

  ICEInteger(FActLock, 1, 0);
  FPool.ActingStart(AActData);
end;

function TActor.Pop(var AActData: PActData): boolean;
begin
  if ICEInteger(FActLock, 0, 1) then
  begin
    EnterCriticalSection(FCS);
    try
      AActData := PActData(Items[0]);
      Delete(0);
      if Count > 0 then
        ICEInteger(FActLock, 1, 0);
    finally
      LeaveCriticalSection(FCS);
    end;
    Result := True;    
  end else
    Result := false;
end;

procedure TActor.Send(ADestActor: TActor; AAct: TAct; var AParam: Variant);
var
  NewAct: PActData;
begin
  New(NewAct);
  NewAct.Act := AAct;
  NewAct.Dest := ADestActor;
  NewAct.Sender := Self;
  NewAct.Data := AParam;

  ADestActor.Queue(Self, NewAct);
end;

procedure TActor.Send(ADestActor: TActor; AAct: TAct;
  AParam: array of Variant);
var
  data: Variant;
begin
  data := VarArrayOf(AParam);
  Send(ADestActor, AAct, data);
end;

procedure TActor.Send(AAct: TAct; var AParam: Variant);
begin
  Send(Self, AAct, AParam);
end;

procedure TActor.Send(AAct: TAct; AParam: array of Variant);
begin
  Send(Self, AAct, AParam);
end;

end.
